package homework1

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object ProcessData {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("processData").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(3L))
    ssc.sparkContext.setLogLevel("warn")

    // 从topic1主题中消费消息
    val topic1 = Array("topic1")
    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe(topic1, getConsumerConf("consumer1"))
    )

    // 向topic2主题生产消息
    val topic2 = "topic2"
    ds.foreachRDD(rdd => {
      rdd.foreachPartition(partition => {
        val producer = ProducerUtils.getProducer
        partition.foreach(x => {
          val line = x.value()
          val items = line.split(",")
          val newLine = items.map(elem => elem.slice(7, elem.indexOf("<", 7))).mkString(" | ")
          val result = producer.send(new ProducerRecord[String, String](topic2, x.key(), newLine)).get()
          println(s"topic:${result.topic()} partition:${result.partition()} offset:${result.offset()}")
        })
      })
    })
    ssc.start()
    ssc.awaitTermination()
  }

  def getConsumerConf(groupId: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux100:9092,linux101:9092,linux102:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )
  }
}
